package com.example.partition;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class SalaryTotalMapper
                         // k1        v1      k2          v2
        extends Mapper<LongWritable, Text, IntWritable, Employee> {

    @Override
    protected void map(LongWritable k1, Text v1, Context context) throws IOException, InterruptedException {

          IntWritable k2 = new IntWritable();//部门编号作为k2
        //private   IntWritable v2 = new IntWritable();//工资作为v2
          Employee emp = new Employee(); //员工对象作为v2
        //1. v1由Text类型转换为String：toString()方法
        // 7369,SMITH,CLERK,7902,1980/12/17,800,,20
        String line = v1.toString();
        //2. 按逗号进行分词：split(“,”)方法
        String[] datas = line.split(",");
        // Employee emp = new Employee();
        emp.setEmpno(Integer.parseInt(datas[0]));
        emp.setEname(datas[1]);
        emp.setJob(datas[2]);
        try {
            emp.setMgr(Integer.parseInt(datas[3]));//部门经理
        }catch (Exception e){
            //由于通过数据文件看到部门经理有可能为空,为空会报异常，在此当经理id为空的时候
            //就设置为一个默认值
            emp.setMgr(-1);
        }
        emp.setHiredate(datas[4]);
        emp.setSal(Integer.parseInt(datas[5]));//工资
        try {
            emp.setComm(Integer.parseInt(datas[6]));
        }catch (Exception e){
            //由于通过数据文件看到奖金有可能为空,为空会报异常，在此当奖金为空的时候
            //就设置为一个默认值
            emp.setComm(0);
        }
        emp.setDeptno(Integer.parseInt(datas[7]));

        // 3. 取出需要的字段：部门号：字段[7]、工资：字段[5]
//        String deptNo = datas[7]; //拿到部门编号
//        k2.set(Integer.parseInt(deptNo));
        //v2.set(Integer.parseInt(datas[5]));
        k2.set(emp.getDeptno());

        // 4. 输出k2, v2：context.write()方法
        // context.write(k2,v2);
        context.write(k2,emp);



    }
}
